package com.common.elk;

import com.common.constant.CommonConstant;
import com.common.util.CommonUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ElkClientFactory {

    private final static Logger logger = LoggerFactory.getLogger(ElkClientFactory.class);
    private volatile static TransportClient client;
    private volatile static IndexRequest source;

    private ElkClientFactory() {
    }

    public static TransportClient getInstance() throws UnknownHostException {
        logger.info("[INFO-ElkClientFactory] getInstance method");
        Properties prop = CommonUtils.getProperty();
        String elkHost = StringUtils.isNoneBlank(prop.getProperty(CommonConstant.ELK_HOST_KEY)) ? prop.getProperty(CommonConstant.ELK_HOST_KEY) : CommonConstant.DEFAULT_ELK_HOST;
        String elkPort = StringUtils.isNoneBlank(prop.getProperty(CommonConstant.ELK_PORT_KEY)) ? prop.getProperty(CommonConstant.ELK_PORT_KEY) : CommonConstant.DEFAULT_ELK_PORT;
        logger.info("[INFO-ElkClientFactory] ELK {0}:{1}--",elkHost,elkPort);
        if (client == null) {
            synchronized (TransportClient.class) {
                if (client == null) {
                    client = new PreBuiltTransportClient(Settings.EMPTY)
                            .addTransportAddress(new TransportAddress(InetAddress.getByName(elkHost), Integer.parseInt(elkPort)));
                }
            }
        }
//        Settings settings = Settings.builder().put("cluster.name", "elasticsearch").build();
//        logger.info("success connect");
//        return client = new PreBuiltTransportClient(settings);
        return client;
    }

    public static String getOperation(TransportClient client) {
        String result = null;
        GetResponse response = client.prepareGet("megacorp", "employee", "1").get();
        logger.info(response.getSourceAsString());
        result = response.getSourceAsString();
        client.close();
        return result;
    }

    /**
     * indexRequest().source(json, XContentType.JSON)
     *
     * @param client
     * @throws IOException
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void merge(TransportClient client, String json, String id) throws IOException, ExecutionException, InterruptedException {
        logger.info("[INFO-ElkClientFactory] enter merge method");
        long start = System.currentTimeMillis();
        Properties prop = CommonUtils.getProperty();
        String index = StringUtils.isNotBlank(prop.getProperty(CommonConstant.ELK_INDEX_KEY)) ? prop.getProperty(CommonConstant.ELK_INDEX_KEY) : CommonConstant.DEFAULT_ELK_INDEX;
        String type = StringUtils.isNotBlank(prop.getProperty(CommonConstant.ELK_TYPE_KEY)) ? prop.getProperty(CommonConstant.ELK_TYPE_KEY) : CommonConstant.DEFAULT_ELK_TYPE;
        IndexRequest indexRequest = new IndexRequest(index, type, id).source(json, XContentType.JSON);

        UpdateRequest updateRequest = new UpdateRequest(index, type, id).doc(json, XContentType.JSON)
                .upsert(indexRequest);
        client.update(updateRequest).get();
        client.close();
        long end = System.currentTimeMillis();
        logger.info("execute finish {} ms",end-start);
    }
}
